package com.fanxuankai.ksql.function.udf.datetime;

import io.confluent.ksql.function.udf.Udf;
import io.confluent.ksql.function.udf.UdfDescription;
import io.confluent.ksql.function.udf.UdfParameter;
import org.joda.time.DateTime;
import org.joda.time.Months;
import org.joda.time.Years;

import java.sql.Timestamp;
import java.util.Date;

/**
 * @author fanxuankai
 */
@UdfDescription(name = "timestamp_diff", description = "计算毫秒数时间差", author = "fanxuankai")
public class TimestampDiff {
    private static final String P_UNIT = "unit";
    private static final String P_FROM = "from";
    private static final String P_TO = "to";

    @Udf
    public Long diff(@UdfParameter(P_UNIT) String unit, @UdfParameter(P_FROM) Timestamp from,
                     @UdfParameter(P_TO) Timestamp to) {
        return Unit.valueOf(unit).diff(from, to);
    }

    @Udf
    public Long diff(@UdfParameter(P_UNIT) String unit, @UdfParameter(P_FROM) java.sql.Date from,
                     @UdfParameter(P_TO) java.sql.Date to) {
        return Unit.valueOf(unit).diff(from, to);
    }

    @Udf
    public Long diff(@UdfParameter(P_UNIT) String unit, @UdfParameter(P_FROM) java.sql.Time from,
                     @UdfParameter(P_TO) java.sql.Time to) {
        return Unit.valueOf(unit).diff(from, to);
    }

    public enum Unit {
        /**
         * 毫秒数
         */
        MILLIS {
            @Override
            public long getMillis() {
                return 1;
            }
        }, SECOND {
            @Override
            public long getMillis() {
                return MILLIS.getMillis() * 1000;
            }
        }, MINUTE {
            @Override
            public long getMillis() {
                return SECOND.getMillis() * 60;
            }
        }, HOUR {
            @Override
            public long getMillis() {
                return MINUTE.getMillis() * 60;
            }
        }, DAY {
            @Override
            public long getMillis() {
                return HOUR.getMillis() * 24;
            }
        }, WEEK {
            @Override
            public long getMillis() {
                return DAY.getMillis() * 7;
            }
        }, MONTH {
            @Override
            public Long diff(Date from, Date to) {
                if (from == null || to == null) {
                    return null;
                }
                return (long) Months.monthsBetween(new DateTime(from), new DateTime(to)).getMonths();
            }
        }, QUARTER {
            @Override
            public Long diff(Date from, Date to) {
                Long diff = MONTH.diff(from, to);
                if (diff == null) {
                    return diff;
                }
                return diff / 3;
            }
        }, YEAR {
            @Override
            public Long diff(Date from, Date to) {
                if (from == null || to == null) {
                    return null;
                }
                return (long) Years.yearsBetween(new DateTime(from), new DateTime(to)).getYears();
            }
        };

        public Long diff(Date from, Date to) {
            if (from == null || to == null) {
                return null;
            }
            return (to.getTime() - from.getTime()) / getMillis();
        }

        public long getMillis() {
            throw new AbstractMethodError();
        }
    }
}
